package com.example.pushMessage;
import cn.hutool.core.lang.TypeReference;
import cn.hutool.json.JSONUtil;
import com.example.entity.Request;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Lazy;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

@Lazy
public abstract class RocketmqPushMessageListener implements RocketMQListener<MessageExt> {
    private Logger logger = LoggerFactory.getLogger(RocketmqPushMessageListener.class);

    private RocketmqHandlerRegistry rocketmqHandlerRegistry;
   // private EtcdTemplate etcdTemplate;

    public RocketmqPushMessageListener(RocketmqHandlerRegistry rocketmqHandlerRegistry) {
        this.rocketmqHandlerRegistry = rocketmqHandlerRegistry;
    }

  /*  @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.etcdTemplate = applicationContext.getBean(EtcdTemplate.class);
    }*/

    @Override
    public void onMessage(MessageExt message) {
        if (message == null) return;

        if (!rocketmqHandlerRegistry.initialized()) {
            logger.info("--------------------------------------------------------------------------");
            throw new RuntimeException("WARNING: ROCKETMQ HANDLER REGISTRY HAS NOT BEEN INITIALIZED");
        }

        String topic = message.getTopic();
        String tag = message.getTags();
        String destination = String.format("%s:%s", topic, tag);
        String json = new String(message.getBody());

        RocketmqMessageHandler handler =
                rocketmqHandlerRegistry.getHandler(destination);
        if (handler == null) {
            logger.info("--------------------------------------------------------------------------");
            logger.warn("destination {} do not supported", destination);
            logger.info(json);
        } else {
            ParameterizedType requestType = rocketmqHandlerRegistry.getRequestType(destination);
            Request request = JSONUtil.toBean(json, new TypeReference<Request>() {
                @Override
                public Type getType() {
                    return requestType;
                }
            }, true);

            //if (handler instanceof RocketmqMessageSerializationHandler) {
       /*         RocketmqMessageSerializationHandler idempotent = (RocketmqMessageSerializationHandler) handler;
                etcdTemplate.serialization(idempotent.getLockName(request), idempotent.getTTL(), () -> {
                    callWithThrow(handler, request);

                });
            } else {
                callWithThrow(handler, request);*/
                callWithThrow(handler, request);
           // }
        }
    }

    private void callWithThrow(RocketmqMessageHandler handler, Request request) {
        try {
            handler.consuming(request);
        } catch (Exception e) {
            logger.info("--------------------------");
            //logger.error(JsonUtil.toJson(request), e);
            throw new RuntimeException(e);
        }
    }
}